package 实训二;


import com.alibaba.fastjson.JSONObject;
import com.bw.app.bean1.DwsPopularityMetricbean;
import com.bw.app.bean1.DwsUserBehaviorbean;
import com.bw.app.bean1.ShopPopularityKeySelector;

import com.bw.app.func.PopularityProcessFunc;
import com.bw.app.func.RedisSink;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;


/**
 * Flink店铺多维度人气指标实时计算
 * Kafka (dwd_traffic_action_log)
 *         ↓
 * JSON 字符串流
 *         ↓
 * FlatMap 解析 + 过滤无效行为
 *         ↓
 * DwsUserBehaviorbean 流（含 shop_id, sku_id, behavior_type 等）
 *         ↓
 * KeyBy(店铺+渠道+行为类型)
 *         ↓
 * ProcessFunction：实时聚合（PV/UV/热度等）
 *         ↓
 * 输出到 Redis + 控制台
 */

public class DwsFlinkShopPopularityCalculator {
    public class BehaviorWithKey {
        public DwsUserBehaviorbean behavior;
        public Tuple4<String,String,String,String> key;

        public BehaviorWithKey(DwsUserBehaviorbean b,
                               Tuple4<String,String,String,String> k) {
            this.behavior = b;
            this.key = k;
        }
    }

    public static void main(String[] args) throws Exception {
        // 1. 初始化Flink执行环境
        final StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 2. 配置Kafka消费者
        DataStreamSource<String> source = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer("dwd_traffic_action_log", "traffic_action_2212A"));

        // 4. 解析JSON数据为DwsUserBehaviorbean对象
        SingleOutputStreamOperator<DwsUserBehaviorbean> dwsBehaviorbean = source.flatMap(new FlatMapFunction<String, DwsUserBehaviorbean>() {
            @Override
            public void flatMap(String s, Collector<DwsUserBehaviorbean> collector) throws Exception {
                //todo 将每条 JSON 字符串解析为 JSONObject。
                JSONObject jsonObject = JSONObject.parseObject(s);
                // TODO: 2025/9/24 提取时间戳 ts 和用户 ID（从 common 嵌套对象中取 uid）。 
                Long ts = jsonObject.getLong("ts");
                String user_id = jsonObject.getJSONObject("common").getString("uid");
                // TODO: 2025/9/24 只有当 action.item_type == "sku_id" 时，才认为这是一个有效的商品行为，才提取 sku_id；
                String sku_id = null;
                // 2. 仅在满足条件时给 sku_id 赋值
                if (jsonObject.getJSONObject("action") != null // 先判空，避免空指针异常
                        && "sku_id".equals(jsonObject.getJSONObject("action").getString("item_type"))) {
                    sku_id = jsonObject.getJSONObject("action").getString("item");
                }
                // TODO: 2025/9/24 提取行为类型（如 click、cart、buy）、渠道 ch、店铺 ID shop_id。
                String behavior_type = jsonObject.getJSONObject("action").getString("action_id");
                String ch = jsonObject.getJSONObject("common").getString("ch");
                String shop_id = jsonObject.getJSONObject("page").getString("shop_id");
                // TODO: 2025/9/24 过滤掉 sku_id 为空的记录（即非商品相关行为，如页面浏览但未关联商品）。
                DwsUserBehaviorbean dwsUserBehaviorbean = new DwsUserBehaviorbean(ts, user_id, sku_id, behavior_type, ch, shop_id);
                collector.collect(dwsUserBehaviorbean);
            }
        }).filter(t->t.getSku_id()!= null);
        dwsBehaviorbean.print();


//        // 5. 多维度分组并计算人气指标（5分钟滚动窗口）
        // TODO: 2025/9/24 聚合计算人气指标按 ShopPopularityKeySelector 定义的维度分组
        DataStream<DwsPopularityMetricbean> popularityStream = dwsBehaviorbean
                .keyBy(new ShopPopularityKeySelector())
                // TODO: 2025/9/24 使用自定义的 PopularityProcessFunc 进行状态管理与指标计算（如 5 分钟窗口内的 PV、UV、去重用户数等）；
                .process(new PopularityProcessFunc());   // ✅ 换这里
        // 6. 将计算结果写入Redis

        popularityStream.addSink(new RedisSink());

        // 7. 输出到控制台用于调试
        popularityStream.print("人气指标: ");

        // 执行Flink作业
        env.execute("DwsFlinkShopPopularityCalculator");
        /***
         * 1. 从 Kafka 消费用户行为日志
         * 主题：dwd_traffic_action_log
         * 日志格式：JSON，包含用户ID、时间戳、行为类型（点击/加购等）、商品ID（sku_id）、店铺ID（shop_id）、渠道（ch）等。
         * 2. 清洗并过滤有效数据
         * 只保留 action.item_type == "sku_id" 的行为（即明确关联到商品的行为）；
         * 提取关键字段：user_id, sku_id, behavior_type, shop_id, ch；
         * 过滤掉 sku_id 为空的无效记录。
         * 3. 按维度分组，实时聚合人气指标
         * 分组维度：由 ShopPopularityKeySelector 定义，通常是：
         * 店铺 ID（shop_id）
         * 渠道（ch）
         * 行为类型（behavior_type）
         * 可能还有时间窗口（如每5分钟）
         * 聚合内容：在 PopularityProcessFunc 中实现，可能包括：
         * PV（页面浏览量）
         * UV（独立用户数）
         * 热度得分等
         * 输出目标：Redis（用于实时展示店铺热度、排行榜等）
         */
    }
}
